import os
import io
import zipfile
from config.path_conf import BASE_DIR

from collections.abc import Sequence
from anyio import open_file
from pydantic.alias_generators import to_pascal
from modules.code_genrator.curd.code_curd import CodeCurd
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import RowMapping
from modules.code_genrator.schemas.gen_business_schema import GenBusinessCreate
from modules.code_genrator.schemas.gen_column_schema import GenColumnCreate
from modules.code_genrator.services.gen_business_service import GenBusinessService
from modules.code_genrator.services.gen_column_service import GenColumnService
from modules.code_genrator.utils.type_conversion import (
    sql_type_to_sqlalchemy,
    sql_type_to_pydantic,
)
from modules.code_genrator.services.gen_business_service import GenBusinessService
from modules.code_genrator.utils.code_template import code_template
from modules.code_genrator.schemas.code_schema import ImportRequest
from utils.log_util import logger


class CodeService:
    """ 代码生成服务 """
    @staticmethod
    async def get_all_tables(*, db: AsyncSession, table_schema: str) -> Sequence[RowMapping]:
        """ 获取数据库所有表 """
        tables = await CodeCurd.get_tables(db=db, table_schema=table_schema)
        return tables
        
    @staticmethod
    async def import_business_and_model(*, db: AsyncSession, import_request: ImportRequest) -> None:
        """ 导入业务代码和模型代码 """
        table_info = await CodeCurd.get_table(db=db, table_schema=import_request.table_schema, table_name=import_request.table_name)
        if not table_info:
            raise ValueError(f"表 {import_request.table_schema}.{import_request.table_name} 不存在")

        # 1) 创建业务记录
        app_name = import_request.app_name
        class_name = to_pascal(import_request.table_name)
        table_name = import_request.table_name
        schema_name = class_name
        filename = import_request.table_name.lower()
        table_comment = table_info["table_comment"] if isinstance(table_info, RowMapping) else table_info[1]

        business_payload = GenBusinessCreate(
            app_name=app_name,
            table_name=table_name,
            table_comment=table_comment,
            class_name=class_name,
            schema_name=schema_name,
            filename=filename,
            default_datetime_column=True,
            api_version="v1",
            gen_path=None,
            remark=None,
        )

        business = await GenBusinessService.create_gen_business(db, business_payload)

        # 2) 拉取列并创建模型字段记录
        columns = await CodeCurd.get_all_columns(db=db, table_schema=import_request.table_schema, table_name=import_request.table_name)
        for col in columns:
            name = col["column_name"]
            is_pk = bool(col["is_pk"])  # 0/1 -> bool
            is_nullable = bool(col["is_nullable"])  # 0/1 -> bool
            sort = col["sort"]
            comment = col["column_comment"]
            column_type = col["column_type"]
            default = col["column_default"]
            column_length = col["column_length"]

            gen_column_payload = GenColumnCreate(
                name=name,
                comment=comment,
                type=sql_type_to_sqlalchemy(column_type),
                pd_type=sql_type_to_pydantic(column_type),
                default=default,
                sort=sort,
                length=int(column_length or 0),
                is_pk=is_pk,
                is_nullable=is_nullable,
                gen_business_id=business.id,
            )
            await GenColumnService.create_gen_column(db, gen_column_payload)

    @staticmethod
    async def render_tpl_code(*, db: AsyncSession, business) -> dict[str, str]:
        """ 渲染模板代码 """
        gen_models = await GenColumnService.get_all_gen_columns_by_business(db, business.id)

        if not gen_models:
            raise ValueError(f"业务记录 {business.id} 不存在")
        gen_vars = code_template.get_vars(business=business, models=gen_models)
        # 返回生成文件路径到渲染内容的映射，便于后续写入或打包
        rendered: dict[str, str] = {}
        for tpl_path in code_template.get_template_files():
            output_path = code_template.get_code_gen_path(business, tpl_path)
            content = await code_template.get_template(tpl_path).render_async(**gen_vars)
            rendered[output_path] = content
        return rendered

    @staticmethod
    async def preview(self, *, db: AsyncSession, pk: int) -> dict[str, bytes]:
        """预览渲染后的模板代码"""
        business = await GenBusinessService.get_gen_business_by_id(db, pk)
        if not business:
            raise ValueError(f"代码生成业务 {pk} 不存在")

        tpl_code_map  = await self.render_tpl_code(db=db, business=business)
        codes = {}
        for tpl_path, content in tpl_code_map.items():
            if tpl_path.startswith("app/modules"):
                codes[tpl_path] = content.encode("utf-8")

    @staticmethod
    async def generate(self, *, db: AsyncSession, pk: int) -> str:
        """生成代码"""
        business = await GenBusinessService.get_gen_business_by_id(db, pk)
        if not business:
            raise ValueError(f"代码生成业务 {pk} 不存在")

        tpl_code_map  = await self.render_tpl_code(db=db, business=business)
        # 将渲染出的相对路径统一转换为绝对磁盘路径（仓库根目录作为基准）
        for tpl_path, content in tpl_code_map.items():
            # 如果模板路径是相对的（以 app/ 开头），将其拼接到项目根目录
            code_filepath = (
                os.path.join(BASE_DIR.parent, tpl_path)
                if not os.path.isabs(tpl_path)
                else tpl_path
            )

            #写入init文件
            code_folder = anyio.Path(code_filepath).parent
            await code_folder.mkdir(parents=True, exist_ok=True)

            init_file = code_folder.joinpath("__init__.py")
            if not await init_file.exists():
                async with await open_file(init_file, "w", encoding="utf-8") as f:
                    await f.write("")
            # routers __init__.py
            if 'routers' in code_filepath:
                router_init = code_folder.joinpath("__init__.py")
                if not await router_init.exists():
                    async with await open_file(router_init, "w", encoding="utf-8") as f:
                        await f.write("")
            # app __init__.py
            if 'service' in code_filepath:
                app_init = code_folder.parent.joinpath("__init__.py")
                if not await app_init.exists():
                    async with await open_file(app_init, "w", encoding="utf-8") as f:
                        await f.write("")
            # models __init__.py
            if 'models' in code_filepath:
                model_init = code_folder.joinpath("__init__.py")
                if not await model_init.exists():
                    async with await open_file(model_init, "w", encoding="utf-8") as f:
                        await f.write("")
            # 写入代码文件
            async with await open_file(code_filepath, "w", encoding="utf-8") as f:
                await f.write(content)

        return code_filepath

    @staticmethod
    async def write_tpl_code(*, db: AsyncSession, business, overwrite: bool = False) -> list[str]:
        """将渲染后的代码写入磁盘

        Args:
            db (AsyncSession): 数据库会话
            business: 业务记录
            overwrite (bool): 是否覆盖已有文件，默认 False
        Returns:
            list[str]: 写入的文件路径列表
        """
        rendered = await CodeService.render_tpl_code(db=db, business=business)
        written: list[str] = []

        # 如果没有可写入的模板，直接返回空列表，避免 StopIteration
        if not rendered:
            logger.warning(f"业务 {business.id} 无模板可生成，已跳过写盘。")
            return written

        # 使用业务信息计算基础目录（绝对路径），避免依赖 rendered 的任意路径
        # 若业务 app_name 包含非 ASCII（如中文），回退到 filename 以保证英文安全目录
        safe_app_name = business.app_name if business.app_name.isascii() else business.filename
        base_dir = os.path.join(BASE_DIR, "modules", safe_app_name)
        subdirs = ["routers", "schemas", "params", "models", "services", "vos", "curd"]

        os.makedirs(base_dir, exist_ok=True)
        # 创建基础 __init__.py
        base_init = os.path.join(base_dir, "__init__.py")
        if not os.path.exists(base_init):
            async with await open_file(base_init, "w", encoding="utf-8") as f:
                await f.write("")

        # 创建各子目录与 __init__.py
        for sd in subdirs:
            sd_path = os.path.join(base_dir, sd)
            os.makedirs(sd_path, exist_ok=True)
            sd_init = os.path.join(sd_path, "__init__.py")
            if not os.path.exists(sd_init):
                async with await open_file(sd_init, "w", encoding="utf-8") as f:
                    await f.write("")

        # 写入各文件
        for path, content in rendered.items():
            # 统一转换为绝对路径（仓库根目录作为基准）
            abs_path = (
                os.path.join(BASE_DIR.parent, path)
                if not os.path.isabs(path)
                else path
            )
            # 如果文件存在且不覆盖，则跳过
            if (not overwrite) and os.path.exists(abs_path):
                continue
            # 确保父目录存在
            os.makedirs(os.path.dirname(abs_path), exist_ok=True)
            async with await open_file(abs_path, "w", encoding="utf-8") as f:
                await f.write(content)
            # 返回依旧用逻辑路径，方便前端展示
            written.append(path)

        return written

    @staticmethod
    async def export_tpl_code_zip(*, db: AsyncSession, business) -> bytes:
        """打包渲染后的代码为 ZIP 文件字节流"""
        rendered = await CodeService.render_tpl_code(db=db, business=business)
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
            for path, content in rendered.items():
                # 使用统一的归档路径分隔符
                arcname = path.replace("\\", "/")
                zf.writestr(arcname, content)
        buffer.seek(0)
        return buffer.getvalue()